package main

import (
	"fmt"
	"github.com/panjf2000/ants/v2"
	"log"
	"math/rand"
	"runtime"
	"sync"
	"time"
)

/**
 * @Description
 * @Author : 弘泰通保
 * @Date 2025/7/18 14:38
 **/

func GoroutineNum() {
	// 1. 统计当前存在的goroutine的数量
	go func() {
		for {
			fmt.Println("goroutineNum: ", runtime.NumGoroutine())
			time.Sleep(500 * time.Microsecond)
		}
	}()

	//启动大量的goroutine
	for {
		go func() {
			time.Sleep(100 * time.Second)
		}()
	}
}

func GoroutineAnts() {
	// 1. 统计当前存在的goroutine的数量
	go func() {
		for {
			fmt.Println("goroutineNum: ", runtime.NumGoroutine())
			time.Sleep(500 * time.Microsecond)
		}
	}()

	// 初始化协程池，goroutine pool NewPool：创建池子
	size := 1024
	pool, err := ants.NewPool(size)
	if err != nil {
		log.Fatal(err)
	}

	// 保证pool被关闭 单个goroutine 的资源占用可能动态增长（栈可扩 缩），需主动释放避免失控。
	defer pool.Release()

	for {
		// 利用pool，调度需要并发的大量goroutine
		// 向pool中提交一个执行的goroutine
		err := pool.Submit(func() {
			v := make([]int, 1024)
			_ = v
			fmt.Println("in goroutine")
			time.Sleep(100 * time.Second)
		})
		if err != nil {
			log.Fatalln(err)
		}
	}
}

// ChannelSync 无缓冲 同步channel
func ChannelSync() {
	wg := sync.WaitGroup{}
	ch := make(chan int)

	// 间隔发送
	wg.Add(1)
	go func() {
		defer wg.Done()
		for i := 0; i < 5; i++ {
			// send to ch
			ch <- i
			//格式化时间
			fmt.Println("send ", i, "\tNow:", time.Now().Format("2006-01-02 15:04:05"))
			//间隔时间
			time.Sleep(1 * time.Second)
		}
		close(ch)
	}()

	// 间隔接收
	wg.Add(1)
	go func() {
		defer wg.Done()
		for value := range ch {
			fmt.Println("Received ", value, "\tNow:", time.Now().Format("2006-01-02 15:04:05"))
			//间隔时间和send不同
			time.Sleep(3 * time.Second)
		}
	}()
	wg.Wait()
}

func ChannelDirectional() {
	// 双向channel
	ch := make(chan int)
	// *sync.WaitGroup 此时是指针类型
	wg := &sync.WaitGroup{}

	// 单向channel函数调用
	wg.Add(2)
	// 使用双向channel为单向channel赋值
	go setElement(ch, 5, wg)
	go getElement(ch, wg)
	wg.Wait()
}

// only receive channel 仅接受channel,发送报错
func getElement(ch <-chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	age := <-ch
	fmt.Println("receive from ch, element is", age)
}

// only send channel 仅发送channel，接收报错
func setElement(ch chan<- int, age int, wg *sync.WaitGroup) {
	defer wg.Done()
	ch <- age
	fmt.Println("send to ch, element is", age)
}

func SelectRace() {

	// 初始化数据
	// 模拟查询结果，需要与具体的querier建立联系
	type Rows struct {
		// 数据字段

		// 索引标识
		Index int
	}

	// 模拟querier查询器的数量
	const QuerierNum = 8

	// 用于通信的channel，数据channel，停止信号channel
	// 缓冲channel一般用来做数据通信
	ch := make(chan Rows, 1)
	// 无缓冲channel 一般用来做数据同步
	stopChs := [QuerierNum]chan struct{}{}
	for i := range stopChs {
		stopChs[i] = make(chan struct{})
	}

	// wg, rand
	wg := sync.WaitGroup{}
	rand.Seed(time.Now().UnixMilli())

	// 模拟querier查询，每个查询持续不同的时间
	wg.Add(QuerierNum)
	for i := 0; i < QuerierNum; i++ {
		// 每一个querier
		go func(i int) {
			defer wg.Done()

			// 模拟执行时间
			ranD := rand.Intn(1000)
			fmt.Println("goroutine", i, "start fetch data,  Need durantion is", ranD, "ms：")
			// 查询结果的channel
			cnRst := make(chan Rows, 1)

			// 执行查询工作
			go func() {
				// 模拟时长
				time.Sleep(time.Duration(ranD) * time.Microsecond)
				cnRst <- Rows{
					Index: i,
				}
			}()

			// 监听查询结果和停止的信号channel
			select {
			// 查询结果
			case rows := <-cnRst:
				fmt.Println("querier ", i, "get result.")
				// 保证没有其他结果写入才写入结果
				if len(ch) == 0 {
					ch <- rows
				}

			// stop信号
			case <-stopChs[i]:
				fmt.Println("querier ", i, "is stopping")
				return
			}
		}(i)
	}

	// 等待第一个查询结果的反馈
	wg.Add(1)
	go func() {
		defer wg.Done()
		// 等待ch中传递的结果
		select {
		// 等待第一个查询结果
		case rows := <-ch:
			fmt.Println("get first result from ", rows.Index, "stop other querier.")
			// 循环结构，全部通知querier结束
			for i := range stopChs {
				// 当前返回结果的 goroutine 不需要了，因为已经结束
				if i == rows.Index {
					continue
				}
				stopChs[i] <- struct{}{}
			}

		//计划一个超时时间
		case <-time.After(5 * time.Second):
			fmt.Println("all querier timeout")
			// 循环结构，全部通知 querier 结束
			for i := range stopChs {
				stopChs[i] <- struct{}{}
			}
		}
	}()

	wg.Wait()
}
